package drds.global_transaction.resource_manager;

import static drds.global_transaction.api.rpc.rpc_client.server_addresss_look_up_service.ConfigurationKeys.client_async_phase2commitcontext_list_max_size;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import drds.global_transaction.api.api.BranchTransactionStatus;
import drds.global_transaction.api.api.ResourceManagerInbound;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.config.ConfigurationFactory;
import drds.global_transaction.resource_manager.proxy.resource_manager.DataSourceProxy;
import drds.global_transaction.resource_manager.undolog.UndoLogManager;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class AsyncPhase2CommitWorker implements ResourceManagerInbound
{
	//
	private static final List<Phase2CommitContext> async_phase2commitcontext_list = Collections.synchronizedList(new ArrayList<Phase2CommitContext>());
	private static int async_phase2commitcontext_list_max_size = ConfigurationFactory.getInstance().getInt(client_async_phase2commitcontext_list_max_size, 10000);
	private static ScheduledExecutorService scheduledExecutorService;

	@Override
	public BranchTransactionStatus branchTransactionCommit(String globalTransactionIdWithAddressAndPort, long branchTransactionId, String resourceId, String applicationData) throws GlobalTransactionException
	{
		if (async_phase2commitcontext_list.size() < async_phase2commitcontext_list_max_size)
		{
			async_phase2commitcontext_list.add(new Phase2CommitContext(globalTransactionIdWithAddressAndPort, branchTransactionId, resourceId, applicationData));
		} else
		{
			log.warn("async global transaction commit buffer is full. rejected branch [" + branchTransactionId + "/" + globalTransactionIdWithAddressAndPort + "] will be handled by housekeeping later.");
		}
		return BranchTransactionStatus.phase_two_committed;
	}

	public synchronized void init()
	{
		log.info("async commit buffer limit: " + async_phase2commitcontext_list_max_size);
		scheduledExecutorService = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("async phase2commit worker", 1, true));
		scheduledExecutorService.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{

					doBranchCommits();

				} catch (Throwable e)
				{
					log.info("failed at async committing ... " + e.getMessage());

				}
			}
		}, 10, 1000 * 1, TimeUnit.MILLISECONDS);
	}

	private void doBranchCommits()
	{
		if (async_phase2commitcontext_list.size() == 0)
		{
			return;
		}
		Map<String, List<Phase2CommitContext>> resourceIdToPhase2CommitContextListMap = new HashMap<>();
		Iterator<Phase2CommitContext> iterator = async_phase2commitcontext_list.iterator();
		while (iterator.hasNext())
		{
			Phase2CommitContext phase2CommitContext = iterator.next();
			List<Phase2CommitContext> phase2CommitContextList = resourceIdToPhase2CommitContextListMap.get(phase2CommitContext.resourceId);
			if (phase2CommitContextList == null)
			{
				phase2CommitContextList = new ArrayList<>();
				resourceIdToPhase2CommitContextListMap.put(phase2CommitContext.resourceId, phase2CommitContextList);
			}
			phase2CommitContextList.add(phase2CommitContext);

			iterator.remove();

		}

		for (String resourceId : resourceIdToPhase2CommitContextListMap.keySet())
		{
			Connection connection = null;
			try
			{
				try
				{
					DataSourceProxy dataSourceProxy = ResourceManager.get().get(resourceId);
					connection = dataSourceProxy.getNativeConnection();
				} catch (SQLException sqle)
				{
					log.warn("failed to get connection for async committing on " + resourceId, sqle);
					continue;
				}

				List<Phase2CommitContext> phase2CommitContextList = resourceIdToPhase2CommitContextListMap.get(resourceId);
				for (Phase2CommitContext phase2CommitContext : phase2CommitContextList)
				{
					try
					{
						UndoLogManager.deleteUndoLog(//
								connection, phase2CommitContext.globalTransactionIdWithAddressAndPort, //
								phase2CommitContext.branchTransactionId//
						);//
					} catch (Exception ex)
					{
						log.warn("failed to delete undolog log [" + phase2CommitContext.branchTransactionId + "/" + phase2CommitContext.globalTransactionIdWithAddressAndPort + "]", ex);
					}
				}

			} finally
			{
				if (connection != null)
				{
					try
					{
						connection.close();
					} catch (SQLException closeEx)
					{
						log.warn("failed to close JDBC resource while deleting undo_log ", closeEx);
					}
				}
			}

		}

	}

	@Override
	public BranchTransactionStatus branchTransactionRollback(String globalTransactionIdWithAddressAndPort, long branchTransactionId, String resourceId, String applicationData) throws GlobalTransactionException
	{
		throw new UnsupportedOperationException();

	}

}
